package com.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class Step {

    protected static void common(StreamExecutionEnvironment env, boolean isAppointUid) {
        DataStreamSource<String> socketSource = env.socketTextStream("172.16.70.13", 12345);
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapFunction = socketSource.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                if (word.equalsIgnoreCase("bug")) {
                    throw new RuntimeException("Bug Coming.");
                }
                return Tuple2.of(word, 1);
            }
        });

        if (isAppointUid) {
            mapFunction.uid("mapFunId");
        }

        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = mapFunction.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return stringIntegerTuple2.f0;
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = tuple2StringKeyedStream.sum(1);

        if (isAppointUid) {
            sum.uid("sumFunId");
        }
        sum.print();
    }

    /**
     * 没有 CheckPoint 的 容错
     * 这里当nc写入bug时，算子模拟异常
     *
     * @param env
     */
    protected static void step01(StreamExecutionEnvironment env) {
        common(env, false);
    }

    /**
     * CheckPoint
     * 不设置UID
     * 当算子异常发生后 会自动重启  并且有状态的算子也会继续计算
     *
     * @param env
     */
    protected static void step02(StreamExecutionEnvironment env) {
        env.enableCheckpointing(5000);
        // CheckPoint 相关配置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // CheckPoint 后端存储
        env.setStateBackend(new FsStateBackend("file:///C:/chk"));

        common(env, false);
    }

    /**
     * 配置 CheckPoint
     * 设置UID
     *
     * @param env
     */
    protected static void step03(StreamExecutionEnvironment env) {
        env.enableCheckpointing(5000);
        // CheckPoint 相关配置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // CheckPoint 后端存储
        env.setStateBackend(new FsStateBackend("file:///C:/chk"));

        common(env, true);
    }


}
